package com.rocketmq.plugins;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import com.jfinal.plugin.IPlugin;

public class RocketMqPushConsumerPlugin implements IPlugin {

    // 通过push模式消费消息，指定消费者组
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConfig.TOPIC);

    @Override
    public boolean start() {
        try {
            // 指定NameServer的地址
            consumer.setNamesrvAddr(MqConfig.ENDPOINT);

            // 订阅这个topic下的所有的消息
            consumer.subscribe(MqConfig.TOPIC, "*");
            // 最大线程数
            consumer.setConsumeThreadMax(20);
            // 最小线程数
            consumer.setConsumeThreadMin(20);
            // 每次从 broker 中拉取的最大个数，默认是32，如果加大需要修改配置文件 
            consumer.setPullBatchSize(32);
            // 消费线程每次处理消息的最大数量
            consumer.setConsumeMessageBatchMaxSize(1);

            // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                // setConsumeMessageBatchMaxSize 就是指 List<MessageExt> msgs 的大小
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.printf("消费消息:%s", new String(msg.getBody()) + "\n");
                    }

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;

    }

    @Override
    public boolean stop() {
        consumer.shutdown();
        return true;
    }

}
